package org.lql.chaos.rocketmq.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import lombok.Getter;
import lombok.Setter;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
import org.lql.chaos.rocketmq.convert.*;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.converter.*;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author: liangqinglong
 * @date: 2025-08-17
 * @description: RocketMQ增强配置类，集成自定义消息转换器和RocketMQTemplate
 **/
@Configuration
public class RocketMQEnhanceConfig {

    /**
     * 统一序列化配置，使用自定义的多种消息转换器
     */
    @Bean
    public RocketMQMessageConverter rocketMQMessageConverter(ObjectMapper objectMapper) {
        return new CustomRocketMQMessageConverter(objectMapper);
    }

    /**
     * 自定义RocketMQMessageConverter，支持多种消息转换器
     */
    public static class CustomRocketMQMessageConverter extends RocketMQMessageConverter {
        private final CompositeMessageConverter compositeMessageConverter;
        
        public CustomRocketMQMessageConverter(ObjectMapper objectMapper) {
            // 创建自定义转换器列表
            List<MessageConverter> converters = new ArrayList<>();
            
            // 添加自定义的多种类型转换器
            converters.add(new MultiTypeMessageConverter(objectMapper));
            converters.add(new JSONMessageConverter(objectMapper));
            converters.add(new ProtobufMessageConvert());
            converters.add(new StringMessageConvert());
            converters.add(new BytesMessageConvert());
            
            // 设置组合消息转换器
            this.compositeMessageConverter = new CompositeMessageConverter(converters);
        }
        
        @Override
        public MessageConverter getMessageConverter() {
            return compositeMessageConverter;
        }
    }

    /**
     * 自定义ObjectMapper配置
     */
    @Bean
    public ObjectMapper customObjectMapper() {
        ObjectMapper objectMapper = new ObjectMapper();
        // 允许序列化java.time.LocalDateTime
        objectMapper.registerModule(new JavaTimeModule());
        // 禁用时间戳，使用java.time.LocalDateTime格式化
        objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
        return objectMapper;
    }

    /**
     * 额外的自定义属性
     */
    @Bean
    @ConfigurationProperties(prefix = "custom.rocketmq")
    public RocketMQEnhanceConfig.ExtraMqProps extraMqProps() {
        return new RocketMQEnhanceConfig.ExtraMqProps();
    }

    @Getter
    @Setter
    public static class ExtraMqProps {
        private Boolean vipChannelEnabled = false;         // 部分网络/云环境需关 VIP
        private Boolean latencyFaultEnable = true;         // 延迟容错（优选健康 Broker）
        private Integer compressThreshold = 4096;          // 压缩阈值（字节）
        private Integer maxMsgSize = 4 * 1024 * 1024;      // 最大消息体
        private String namespace;                          // 可选：逻辑命名空间
        private String unitName;                           // 可选：单元化部署标记
        private Boolean enableMsgTrace = true;             // 启用轨迹
        private String traceTopic = "RMQ_SYS_TRACE_TOPIC"; // 轨迹 Topic
        private Integer sendTimeoutMs = 3000;
        private Integer retrySync = 2;
        private Integer retryAsync = 2;
        private Boolean retryAnotherBroker = true;
    }

    /**
     * 公共构建器：普通 Producer
     */
    private DefaultMQProducer buildProducer(String group,
                                            RocketMQProperties props,
                                            ExtraMqProps extra) throws MQClientException {
        DefaultMQProducer p = new DefaultMQProducer(group);
        p.setNamesrvAddr(props.getNameServer());
        p.setSendMsgTimeout(extra.getSendTimeoutMs());
        p.setRetryTimesWhenSendFailed(extra.getRetrySync());
        p.setRetryTimesWhenSendAsyncFailed(extra.getRetryAsync());
        p.setRetryAnotherBrokerWhenNotStoreOK(Boolean.TRUE.equals(extra.getRetryAnotherBroker()));
        p.setCompressMsgBodyOverHowmuch(extra.getCompressThreshold());
        p.setMaxMessageSize(extra.getMaxMsgSize());
        p.setVipChannelEnabled(Boolean.TRUE.equals(extra.getVipChannelEnabled()));
        p.setSendLatencyFaultEnable(Boolean.TRUE.equals(extra.getLatencyFaultEnable()));

        // 可选：设置 namespace 与 unitName（做多集群/单元化隔离）
        if (extra.getNamespace() != null) {
            p.setNamespace(extra.getNamespace());
        }
        if (extra.getUnitName() != null) {
            p.setUnitName(extra.getUnitName());
        }

        // 注意：不要在这里启动Producer，RocketMQTemplate会在初始化时自动启动
        // p.start();
        return p;
    }

    /**
     * 公共构建器：事务 Producer
     */
    private TransactionMQProducer buildTxProducer(String group,
                                                  RocketMQProperties props,
                                                  ExtraMqProps extra) throws MQClientException {
        TransactionMQProducer p = new TransactionMQProducer(group);
        p.setNamesrvAddr(props.getNameServer());
        p.setSendMsgTimeout(extra.getSendTimeoutMs());
        p.setRetryTimesWhenSendFailed(extra.getRetrySync());
        p.setRetryTimesWhenSendAsyncFailed(extra.getRetryAsync());
        p.setRetryAnotherBrokerWhenNotStoreOK(Boolean.TRUE.equals(extra.getRetryAnotherBroker()));
        p.setCompressMsgBodyOverHowmuch(extra.getCompressThreshold());
        p.setMaxMessageSize(extra.getMaxMsgSize());
        p.setVipChannelEnabled(Boolean.TRUE.equals(extra.getVipChannelEnabled()));
        p.setSendLatencyFaultEnable(Boolean.TRUE.equals(extra.getLatencyFaultEnable()));

        // 可选：设置 namespace 与 unitName（做多集群/单元化隔离）
        if (extra.getNamespace() != null) {
            p.setNamespace(extra.getNamespace());
        }
        if (extra.getUnitName() != null) {
            p.setUnitName(extra.getUnitName());
        }

        // 为事务消息配置专用线程池
        ExecutorService executorService = new ThreadPoolExecutor(
                5, // 核心线程数
                20, // 最大线程数
                60, // 空闲线程存活时间
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(1000), // 队列大小
                r -> {
                    Thread thread = new Thread(r);
                    thread.setName("Transaction-Check-Thread-" + thread.getId());
                    return thread;
                }
        );
        p.setExecutorService(executorService);

        // 注意：不要在这里启动Producer，RocketMQTemplate会在初始化时自动启动
        // p.start();
        return p;
    }

    /**
     * 订单业务 Producer（高实时）
     */
    @Bean("orderRocketMQTemplate")
    public RocketMQTemplate orderRocketMQTemplate(RocketMQProperties props,
                                                  ExtraMqProps extra,
                                                  RocketMQMessageConverter converter) throws MQClientException {
        DefaultMQProducer producer = buildProducer("order-producer-group", props, extra);

        // 订单业务：高实时性，可容忍一定失败
        producer.setSendMsgTimeout(1500); // 较短超时
        producer.setRetryTimesWhenSendFailed(1); // 较少重试
        producer.setRetryTimesWhenSendAsyncFailed(1);

        RocketMQTemplate template = new RocketMQTemplate();
        template.setProducer(producer);
        template.setMessageConverter(converter.getMessageConverter());
        return template;
    }

    /**
     * 支付业务 Producer（高可靠）
     */
    @Bean("payRocketMQTemplate")
    public RocketMQTemplate payRocketMQTemplate(RocketMQProperties props,
                                                ExtraMqProps extra,
                                                RocketMQMessageConverter converter) throws MQClientException {
        DefaultMQProducer producer = buildProducer("pay-producer-group", props, extra);

        // 更保守：更长超时、更多重试
        producer.setSendMsgTimeout(8000);
        producer.setRetryTimesWhenSendFailed(3);
        producer.setRetryAnotherBrokerWhenNotStoreOK(true);

        RocketMQTemplate template = new RocketMQTemplate();
        template.setProducer(producer);
        template.setMessageConverter(converter.getMessageConverter());
        return template;
    }

    /**
     * 事务消息 Producer
     */
    @Bean("txRocketMQTemplate")
    public RocketMQTemplate txRocketMQTemplate(RocketMQProperties props,
                                               ExtraMqProps extra,
                                               RocketMQMessageConverter converter) throws MQClientException {
        TransactionMQProducer producer = buildTxProducer("tx-producer-group", props, extra);

        RocketMQTemplate template = new RocketMQTemplate();
        template.setProducer(producer); // 事务发送需要 TransactionMQProducer
        template.setMessageConverter(converter.getMessageConverter());
        return template;
    }

    /**
     * 延时消息 Producer
     */
    @Bean("delayRocketMQTemplate")
    public RocketMQTemplate delayRocketMQTemplate(RocketMQProperties props,
                                                  ExtraMqProps extra,
                                                  RocketMQMessageConverter converter) throws MQClientException {
        DefaultMQProducer producer = buildProducer("delay-producer-group", props, extra);

        RocketMQTemplate template = new RocketMQTemplate();
        template.setProducer(producer);
        template.setMessageConverter(converter.getMessageConverter());
        return template;
    }

    /**
     * 异步消息 Producer
     */
    @Bean("asyncRocketMQTemplate")
    public RocketMQTemplate asyncRocketMQTemplate(RocketMQProperties props,
                                                  ExtraMqProps extra,
                                                  RocketMQMessageConverter converter) throws MQClientException {
        DefaultMQProducer producer = buildProducer("async-producer-group", props, extra);
        
        // 异步消息配置
        producer.setRetryTimesWhenSendAsyncFailed(3); // 异步发送失败重试次数

        RocketMQTemplate template = new RocketMQTemplate();
        template.setProducer(producer);
        template.setMessageConverter(converter.getMessageConverter());
        return template;
    }
}